#include "app_pool.h"

static pthread_t *thread_pool;

static mqd_t mq;

static int thread_pool_num;

void *thread_func(void *arg)
{
    // 1 每个线程主动的去争抢任务
    while (1)
    {
        task_t task;
        mq_receive(mq, (char *)&task, sizeof(task), NULL);

        if (task.func != NULL)
        {
            task.func(task.arg);
        }
    }
}

/**
 * 线程池初始化
 */
int app_pool_init(int thread_num)
{
    thread_pool_num = thread_num;
    // 1 初始化队列
    // 1.1 设定队列属性
    struct mq_attr attr;
    attr.mq_maxmsg = MQUEUE_MAXMSG;
    attr.mq_msgsize = sizeof(task_t);

    // 1.2 创建队列
    mq = mq_open(MQUEUE_NAME, O_RDWR | O_CREAT, 0644, &attr);
    if (mq == -1)
    {
        perror("mq_open error");
        return -1;
    }

    // 2 初始化线程
    thread_pool = (pthread_t *)malloc(sizeof(pthread_t) * thread_num);
    for (int i = 0; i < thread_num; i++)
    {
        pthread_create(&thread_pool[i], NULL, (void *)thread_func, NULL);
    }
    return 0;
}

/**
 * 向线程池中添加任务
 */
int app_pool_register_task(void *(*task_func)(void *), void *arg)
{
    // 把需求放的任务定义和参数，组合成为一个任务结构体
    task_t task;
    task.func = task_func;
    task.arg = arg;

    // 把任务结构体放入队列中
    int ret = mq_send(mq, (char *)&task, sizeof(task), 0);
    if (ret == -1)
    {
        perror("mq_send error");
        return -1;
    }
    return 0;
}

void app_pool_destroy(void)
{
    // 1 销毁n线程
    for (int i = 0; i < thread_pool_num ; i++)
    {
       pthread_cancel(thread_pool[i]);
       pthread_join(thread_pool[i],NULL);
    }

    mq_close(mq);
    mq_unlink(MQUEUE_NAME);
}
